package cn.xnatural.enet.event;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static java.util.Collections.emptyList;

/**
 * event publisher 事件发布器.事件分发中心
 * TODO 事件死锁. 事件执行链
 */
public class EP {
    protected Logger                      log;
    protected Executor                    exec;
    /**
     * 事件名 -> 监听器{@link Listener}
     */
    protected Map<String, List<Listener>> lsMap;
    /**
     * 需要追踪执行的事件名字
     */
    protected Set<String>                 trackEvents;


    public EP() { init(null, null); }
    public EP(Executor exec) { init(exec, null); }
    public EP(Executor exec, Logger log) { init(exec, log); }


    /**
     * 初始化
     * @param exec 线程池
     */
    protected void init(Executor exec, Logger log) {
        this.exec = exec;
        if (log == null) this.log = LoggerFactory.getLogger(EP.class);
        else this.log = log;
        lsMap = new ConcurrentHashMap<>();
        trackEvents = ConcurrentHashMap.newKeySet(7);
    }


    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     */
    public Object fire(String eName) { return fire(new EC(eName)); }
    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     */
    public Object fireDebug(String eName) { return fire(new EC(eName).debug()); }

    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param args 监听器方法的参数列表
     */
    public Object fire(String eName, Object...args) { return fire(new EC(eName).args(args)); }
    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param args 监听器方法的参数列表
     */
    public Object fireDebug(String eName, Object...args) { return fire(new EC(eName).args(args).debug()); }

    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param completeFn 事件结束回调
     * @param args 监听器方法的参数列表
     */
    public Object fire(String eName, Consumer<EC> completeFn, Object...args) { return fire(new EC(eName).completeFn(completeFn).args(args)); }
    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param completeFn 事件结束回调
     * @param args 监听器方法的参数列表
     */
    public Object fireDebug(String eName, Consumer<EC> completeFn, Object...args) { return fire(new EC(eName).completeFn(completeFn).args(args).debug()); }


    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     */
    public Object fireSync(String eName) { return fire(new EC(eName).sync()); }
    /**
     * {@link #fire(EC)}
     * @param eName 事件名
     */
    public Object fireSyncDebug(String eName) { return fire(new EC(eName).sync().debug()); }

    /**
     * 强制同步
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param args 监听器方法的参数列表
     */
    public Object fireSync(String eName, Object...args) { return fire(new EC(eName).sync().args(args)); }
    /**
     * 强制同步
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param args 监听器方法的参数列表
     */
    public Object fireSyncDebug(String eName, Object...args) { return fire(new EC(eName).sync().args(args).debug()); }


    /**
     * 强制异步
     * {@link #fire(EC)}
     * @param eName 事件名
     */
    public EP fireAsync(String eName) {
        fire(new EC(eName).async(true));
        return this;
    }
    /**
     * 强制异步
     * {@link #fire(EC)}
     * @param eName 事件名
     */
    public EP fireAsyncDebug(String eName) {
        fire(new EC(eName).async(true).debug());
        return this;
    }

    /**
     * 强制异步
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param args 监听器方法的参数列表
     */
    public EP fireAsync(String eName, Object...args) {
        fire(new EC(eName).async(true).args(args));
        return this;
    }
    /**
     * 强制异步
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param args 监听器方法的参数列表
     */
    public EP fireAsyncDebug(String eName, Object...args) {
        fire(new EC(eName).async(true).args(args).debug());
        return this;
    }

    /**
     * 强制异步
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param completeFn 事件结束回调
     * @param args 监听器方法的参数列表
     */
    public EP fireAsync(String eName, Consumer<EC> completeFn, Object...args) {
        fire(new EC(eName).async(true).completeFn(completeFn).args(args));
        return this;
    }
    /**
     * 强制异步
     * {@link #fire(EC)}
     * @param eName 事件名
     * @param completeFn 事件结束回调
     * @param args 监听器方法的参数列表
     */
    public EP fireAsyncDebug(String eName, Consumer<EC> completeFn, Object...args) {
        fire(new EC(eName).async(true).completeFn(completeFn).args(args).debug());
        return this;
    }

    public Object fire(EC ec) {
        List<Listener> ls = lsMap.get(ec.eName); // 获取需要执行的监听器
        ls = new ArrayList<>(ls == null ? emptyList() : ls); // 避免在执行的过程中动态增删事件的监听器而导致个数错误
        return fire(ec, ls);
    }

    public Object fire(EC ec, Listener... ls) {
        return fire(ec, ls == null || ls.length < 1 ? null : Arrays.asList(ls));
    }

    /**
     * 发布事件到各个监听者并执行
     * @param ec {@link EC} 事件执行过程上下文
     * @return 事件执行结果. Note: 取返回值时, 要注意是同步执行还是异步执行
     */
    public Object fire(EC ec, List<Listener> ls) {
        if (ec == null) throw new NullPointerException("Param ec required");
        if (trackEvents.contains(ec.eName) || log.isTraceEnabled()) ec.track = true;

        ec.start(ls, this); // 初始化

        if (ec.isNoListener()) { ec.tryFinish(); return ec.result(); } // 没有监听器的情况

        if (exec == null) { // 只能同步执行
            for (Listener l : ls) l.invoke(ec);
        } else {
            // 异步, 同步执行的监听器, 分开执行
            List<Listener> asyncLs = new ArrayList<>(ls.size()); // 异步执行的监听器
            List<Listener> syncLs = new ArrayList<>(ls.size()); // 同步执行的监听器
            if (ec.async == null) {
                for (Listener l : ls) {
                    if (l.async) asyncLs.add(l);
                    else syncLs.add(l);
                }
            } else {
                if (ec.async) asyncLs.addAll(ls); // 强制全部异步执行
                else syncLs.addAll(ls); // 强制全部同步执行
            }

            syncLs.forEach(l -> l.invoke(ec)); // 先执行同步监听器
            if (asyncLs.size() > 1) { // 至少两个以上 需要 分优先级 依次执行
                //1. 按优先级分组
                final Map<Float, Queue<Runnable>> orderGroupMap = new TreeMap<>();
                for (Listener listener : asyncLs) {
                    final Queue<Runnable> groupLs = orderGroupMap.computeIfAbsent(listener.order, f -> new ConcurrentLinkedQueue<>());
                    groupLs.add(new Runnable() {
                        @Override
                        public void run() {
                            listener.invoke(ec);
                            groupLs.remove(this);
                            if (!groupLs.isEmpty()) return; // TODO 是否会有多个线程同时执行到下边?
                            if (!orderGroupMap.containsKey(listener.order)) return;
                            synchronized (orderGroupMap) {
                                if (!orderGroupMap.containsKey(listener.order)) return;
                                // 当前分组已执行完
                                orderGroupMap.remove(listener.order);
                                Iterator<Map.Entry<Float, Queue<Runnable>>> it = orderGroupMap.entrySet().iterator();
                                if (it.hasNext()) { // 继续执行下一个分组
                                    it.next().getValue().forEach(exec::execute);
                                }
                            }
                        }
                    });
                }
                //2. 开始执行第一个分组
                orderGroupMap.entrySet().iterator().next().getValue().forEach(exec::execute);
            } else {
                asyncLs.forEach(l -> exec.execute(() -> l.invoke(ec)));
            }
        }
        return ec.result();
    }


    /**
     * 是否存在事件监听器
     * @param eNames 事件名
     * @return true if exist
     */
    public boolean exist(String...eNames) {
        if (eNames == null) return false;
        for (String n : eNames) {
            if (!lsMap.containsKey(n)) return false;
        }
        return true;
    }


    /**
     * 所有事件名
     */
    public Set<String> events() {
        return lsMap.keySet();
    }


    /**
     * 添加对象源(解析出监听器)
     * @param source 事件源
     */
    public EP addListenerSource(Object source) { resolve(source); return this; }


    /**
     * 设置某个事件需要追踪执行
     * @param eNames 事件名字(可以多个)
     */
    public EP addTrackEvent(String... eNames) {
        if (eNames == null) return this;
        for (String n : eNames) if (n != null && !n.trim().isEmpty()) trackEvents.add(n.trim());
        return this;
    }


    /**
     * 删除事件追踪
     * @param eNames 事件名
     */
    public EP delTrackEvent(String... eNames) {
        if (eNames == null) return this;
        for (String n : eNames) if (n != null && !n.trim().isEmpty()) trackEvents.remove(n);
        return this;
    }


    /**
     * 只删除指定对象源中的事件监听
     * @param eName 事件名
     * @param source 监听器对象源
     * @return {@link EP}
     */
    public EP removeEvent(String eName, Object source) {
        log.debug("remove event '{}' with source: {}", eName, source);
        lsMap.get(eName).removeAll(
                lsMap.get(eName).stream().filter(l -> {
                    if (source == null) return true;
                    else if (l instanceof Listener.MethodListener && ((Listener.MethodListener) l).source == source) return true;
                    return false;
                }).collect(Collectors.toSet())
        );
        return this;
    }

    /**
     * 移除事件对应的所有监听器
     * @param eName 事件名
     * @return {@link EP}
     */
    public EP removeEvent(String eName) {
        lsMap.remove(eName);
        return this;
    }


    /**
     * 移除对象源的所有监听器
     * @param source 监听器对象源
     * @return {@link EP}
     */
    public EP removeSourceAllEvent(Object source) {
        for (Iterator<Map.Entry<String, List<Listener>>> it = lsMap.entrySet().iterator(); it.hasNext(); ) {
            List<Listener> ls = it.next().getValue();
            ls.removeIf(l -> l instanceof Listener.MethodListener && ((Listener.MethodListener) l).source == source);
            if (ls.isEmpty()) it.remove();
        }
        return this;
    }


    /**
     * 添加监听
     * @param eName 事件名
     * @param async 是否异步
     * @param order 执行顺序, 越小越先执行
     * @param limit 执行次数限制. <=0 不限制
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, boolean async, float order, int limit, Runnable fn) {
        if (fn == null) throw new NullPointerException("Param fn required");
        List<Listener> ls = getListeners(eName);
        Listener l = new Listener.RunnableListener(fn);
        l.async = async; l.order = order; l.limit(limit);
        ls.add(l); sort(ls);
        return this;
    }


    /**
     * 添加监听
     * @param eName 事件名
     * @param async 是否异步
     * @param order 执行顺序, 越小越先执行
     * @param limit 执行次数限制 <=0 不限制
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, boolean async, float order, int limit, Function fn) {
        if (fn == null) throw new NullPointerException("fn must not be null");
        List<Listener> ls = getListeners(eName);
        Listener l = new Listener.FunctionListener<>(fn);
        l.async = async; l.order = order; l.limit(limit);
        ls.add(l); sort(ls);
        return this;
    }


    /**
     * 添加监听
     * @param eName 事件名
     * @param async 是否异步
     * @param order 执行顺序, 越小越先执行
     * @param limit 执行次数限制. <=0 不限制
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, boolean async, float order, int limit, BiFunction fn) {
        if (fn == null) throw new NullPointerException("fn must not be null");
        List<Listener> ls = getListeners(eName);
        Listener l = new Listener.BiFunctionListener<>(fn);
        l.async = async; l.order = order; l.limit(limit);
        ls.add(l); sort(ls);
        return this;
    }


    /**
     * 添加监听
     * @param eName 事件名
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, Runnable fn) { return listen(eName, false, 0, 0, fn); }

    /**
     * 添加监听
     * @param eName 事件名
     * @param async 是否异步
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, boolean async, Runnable fn) { return listen(eName, async, 0, 0, fn); }

    /**
     * 添加监听
     * @param eName 事件名
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, Function fn) { return listen(eName, false, 0, 0, fn); }

    /**
     * 添加监听
     * @param eName 事件名
     * @param async 是否异步
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, boolean async, Function fn) { return listen(eName, async, 0, 0, fn); }

    /**
     * 添加监听
     * @param eName 事件名
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, BiFunction fn) { return listen(eName, false, 0, 0, fn); }

    /**
     * 添加监听
     * @param eName 事件名
     * @param async 是否异步
     * @param fn 函数
     * @return {@link EP}
     */
    public EP listen(String eName, boolean async, BiFunction fn) { return listen(eName, async, 0, 0, fn); }


    /**
     * 从一个对象中 解析出 所有带有 {@link EL}注解的方法 转换成监听器{@link Listener}
     * 如果带有注解 {@link EL}的方法被重写, 则用子类的方法
     * @param source 事件源
     */
    protected void resolve(final Object source) {
        if (source == null) return;
        iterateMethod(source.getClass(), m -> { // 查找包含@EL注解的方法
            EL el = m.getDeclaredAnnotation(EL.class);
            if (el == null) return;
            for (String n : el.name()) {
                String eName = parseName(n, source);
                if (eName == null) continue;
                Listener.MethodListener listener = new Listener.MethodListener(source, m);
                listener.async = el.async(); listener.order = el.order(); listener.limit(el.limit());
                m.setAccessible(true);

                List<Listener> ls = getListeners(eName);
                // 暂不支持重载方法
                if (ls.stream().filter(l -> l instanceof Listener.MethodListener).map(l -> (Listener.MethodListener)l).anyMatch(l -> l.source == source && Objects.equals(l.m.getName(), listener.m.getName()))) {
                    log.warn("Same source same method name only one listener. source: {}, methodName: {}", source, m.getName());
                    continue;
                }
                log.debug("Add listener [name: {}, source: {}, method: {}, async: {}, order: {}]", eName, source, m.getName(), listener.async, listener.order);
                ls.add(listener); sort(ls);
            }
        });
    }


    /**
     * 监听器排序
     */
    protected void sort(List<Listener> ls) {
        ls.sort((o1, o2) -> {
            if (o1.async && o2.async) return Float.compare(o1.order, o2.order);
            else if (!o1.async && o2.async) return -1;
            else if (o1.async && !o2.async) return 1;
            else return Float.compare(o1.order, o2.order);
        });
    }


    /**
     * 获取 或 创建 监听器
     * @param eName 事件名
     */
    protected List<Listener> getListeners(String eName) {
        List<Listener> ls = lsMap.get(eName);
        if (ls == null) {
            synchronized (this) {
                ls = lsMap.get(eName);
                if (ls == null) {
                    ls = new CopyOnWriteArrayList<>(); lsMap.put(eName, ls);
                }
            }
        }
        return ls;
    }


    protected Pattern p = Pattern.compile("\\{(?<attr>\\w+)\\}");
    /**
     * 支持表达式 {attr}.eventName, {attr}会被替换成 对象中的属性attr的值
     * @param name 事件名
     * @param source 事件源
     * @return 解析后的事件名
     */
    protected String parseName(String name, Object source) {
        Matcher matcher = p.matcher(name);
        if (!matcher.find()) return name;
        String attr = matcher.group("attr");
        // 查找属性对应的get方法
        Method m = findMethod(source.getClass(), mm -> {
            if (Modifier.isPublic(mm.getModifiers()) &&
                mm.getParameterCount() == 0 && // 参数个数为0
                String.class.equals(mm.getReturnType()) && // 返回String类型
                ("get" + (attr.substring(0, 1).toUpperCase() + attr.substring(1))).equals(mm.getName()) // get方法名相同
            ) return true;
            return false;
        });
        if (m != null) {
            try {
                Object v = m.invoke(source);
                if (v == null || v.toString().isEmpty()) {
                    log.warn("Parse event name '{}' error. Get property '{}' is empty from '{}'.", name, attr, source);
                    return null;
                }
                return matcher.replaceAll(v.toString());
            } catch (Exception e) { log.error("", e); }
        }
        log.warn("Parse event name '{}'. Not found property '{}' from {} ", name, attr, source);
        return null;
    }


    // 遍历一个类的方法
    protected void iterateMethod(final Class clz, Consumer<Method>... fns) {
        if (fns == null || fns.length < 1) return;
        Class c = clz;
        do {
            for (Method m : c.getDeclaredMethods()) for (Consumer<Method> fn : fns) fn.accept(m);
            c = c.getSuperclass();
        } while (c != null);
    }


    // 查找类中的方法
    protected Method findMethod(final Class clz, Predicate<Method> predicate) {
        Class c = clz;
        do {
            for (Method m : c.getDeclaredMethods()) if (predicate.test(m)) return m;
            c = c.getSuperclass();
        } while (c != null);
        return null;
    }
}